package com.golxen.framework.netty.session.agent;


import com.golxen.framework.PBPacket.PBPacket;
import com.golxen.utils.LogUtil;
import com.mysql.cj.util.LogUtils;
import io.netty.channel.Channel;
import io.netty.channel.ChannelPromise;
import io.netty.util.concurrent.EventExecutor;
import io.netty.util.concurrent.Future;
import io.netty.util.concurrent.GenericFutureListener;
import org.slf4j.Logger;

import java.util.ArrayList;
import java.util.Queue;
import java.util.concurrent.ConcurrentLinkedDeque;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;

public class AgentChannel {

	private static Logger logger = LogUtil.COMMON;

	private static final int PROMISE_MAX_COUNT = 128;
	private static final int PROMISE_MIN_COUNT = 64;

	public Channel channel;
	private GenericFutureListener<Future<? super Void>> listener;

	private AtomicInteger counter;
	private boolean isWriteTerminal;
	private Queue<PBPacket> bufferQueue;

	private Runnable flushTask;
	private AtomicBoolean flushTaskSwitch;

	public AgentChannel(Channel channel) {
		this.channel = channel;
		counter = new AtomicInteger();
		bufferQueue = new ConcurrentLinkedQueue<>();
		flushTask = new FlushTask();
		flushTaskSwitch = new AtomicBoolean(true);
		listener = future -> {
			counter.decrementAndGet();
			if (freshAndGetWriteTerminal()) {
				flush();
			}
		};
	}

	public void write(PBPacket msg) {
		if (msg == null) {
			return;
		}
		bufferQueue.offer(msg);
		if (freshAndGetWriteTerminal()) {
			flush();
		}
	}

	public void flush() {
		if (bufferQueue.isEmpty()) {
			return;
		}

		if (flushTaskSwitch.getAndSet(false)) {
			EventExecutor executor = channel.eventLoop();
			if (executor.isTerminated()) {
				return;
			}
			executor.execute(flushTask);
		}
	}

	private boolean freshAndGetWriteTerminal() {
		if (isWriteTerminal) {
			if (counter.get() >= PROMISE_MAX_COUNT) {
				isWriteTerminal = false;
			}
		} else {
			if (counter.get() <= PROMISE_MIN_COUNT) {
				isWriteTerminal = true;
			}
		}
		return isWriteTerminal;
	}

	public boolean isAlive() {
		try {
			if (channel == null) {
				return false;
			}
			return channel.isActive();
		} catch (Exception e) {
			logger.error("检查连接状态出现异常", e);
			return false;
		}
	}

	public void close() {
		try {
			LogUtil.COMMON.debug(" idle time out call close cid={}", channel.id().asShortText());
			channel.close();
		} catch (Exception e) {
			logger.error("", e);
		}
	}

	private class FlushTask implements Runnable {

		@Override
		public void run() {
			flushTaskSwitch.set(true);
			if (bufferQueue.size() == 1) {
				PBPacket msg = bufferQueue.poll();
				if (msg != null) {
					channel.writeAndFlush(msg, newPromise());
				}
			} else if (bufferQueue.size() > 1) {
				ArrayList<PBPacket> buffer = new ArrayList<>();
				PBPacket msg;
				while ((msg = bufferQueue.poll()) != null) {
                    buffer.add(msg);
                }
				if (!buffer.isEmpty()) {
					channel.writeAndFlush(new PBPacket(buffer), newPromise());
				}
			} else {
				flush();
			}
		}

		private ChannelPromise newPromise() {
			ChannelPromise promise = channel.newPromise();
			promise.addListener(listener);
			counter.incrementAndGet();
			return promise;
		}
	}

	public Channel getChannel() {
		return channel;
	}

}
